package io.task.message.socket.spi.server;

import io.task.message.socket.ChannelContext;
import io.task.message.socket.ChannelContext.CloseCode;
import io.task.message.socket.Config;
import io.task.message.socket.Io;
import io.task.message.socket.lock.SetWithLock;
import io.task.message.socket.spi.server.intf.ServerHandler;
import io.task.message.socket.task.Handler;
import io.task.message.socket.thread.pool.SynThreadPoolExecutor;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.Set;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;

@Slf4j
@Setter
@Getter
public class ServerConfig extends Config {

    private AcceptCompletionHandler acceptCompletionHandler = null;
    private ServerHandler serverHandler = null;

    private Thread checkHeartbeatThread = null;
    private boolean needCheckHeartbeat = true;

    private boolean isShared = false;


    public ServerConfig(ServerHandler serverHandler) {
        this(null, serverHandler);
    }

    public ServerConfig(String name, ServerHandler serverHandler) {
        this(name, serverHandler, null, null);
    }


    public ServerConfig(ServerHandler serverHandler, SynThreadPoolExecutor tioExecutor, ThreadPoolExecutor groupExecutor) {
        this(null, serverHandler, tioExecutor, groupExecutor);
    }

    public ServerConfig(String name, ServerHandler serverHandler, SynThreadPoolExecutor tioExecutor,
                        ThreadPoolExecutor groupExecutor) {
        super(tioExecutor, groupExecutor);

        init(name, serverHandler, tioExecutor, groupExecutor);
    }


    private void init(String name, ServerHandler tioServerHandler, SynThreadPoolExecutor tioExecutor, ThreadPoolExecutor groupExecutor) {
        this.name = name;

        this.acceptCompletionHandler = new AcceptCompletionHandler();
        this.serverHandler = tioServerHandler;

        checkHeartbeatThread = new Thread(new Runnable() {
            @Override
            public void run() {
                //第一次先休息一下
                try {
                    Thread.sleep(1000 * 10);
                } catch (InterruptedException e1) {
                    log.error(e1.toString(), e1);
                }

                while (needCheckHeartbeat && !isStopped()) {
                    //					long sleeptime = heartbeatTimeout;
                    if (heartbeatTimeout <= 0) {
                        log.info("{}, 用户取消了框架层面的心跳检测，如果业务需要，请用户自己去完成心跳检测", ServerConfig.this.name);
                        break;
                    }
                    try {
                        Thread.sleep(heartbeatTimeout);
                    } catch (InterruptedException e1) {
                        log.error(e1.toString(), e1);
                    }
                    long start = System.currentTimeMillis();
                    SetWithLock<ChannelContext> setWithLock = ServerConfig.this.connections;
                    Set<ChannelContext> set = null;
                    long start1 = 0;
                    int count = 0;
                    ReadLock readLock = setWithLock.readLock();
                    readLock.lock();
                    try {
                        start1 = System.currentTimeMillis();
                        set = setWithLock.getObj();

                        for (ChannelContext channelContext : set) {
                            count++;
                            long compareTime = Math.max(channelContext.stat.latestTimeOfReceivedByte, channelContext.stat.latestTimeOfSentPacket);
                            long currtime = System.currentTimeMillis();
                            long interval = currtime - compareTime;

                            boolean needRemove = false;
                            if (channelContext.heartbeatTimeout != null && channelContext.heartbeatTimeout > 0) {
                                needRemove = interval > channelContext.heartbeatTimeout;
                            } else {
                                needRemove = interval > heartbeatTimeout;
                            }

                            if (needRemove) {
                                if (!onHeartbeatTimeout(channelContext, interval, channelContext.stat.heartbeatTimeoutCount.incrementAndGet())) {
                                    log.info("{}, {} ms没有收发消息", channelContext, interval);
                                    channelContext.setCloseCode(CloseCode.HEARTBEAT_TIMEOUT);
                                    Io.remove(channelContext, interval + " ms没有收发消息");
                                }
                            }
                        }
                    } catch (Throwable e) {
                        log.error("", e);
                    } finally {
                        try {
                            readLock.unlock();

                        } catch (Throwable e) {
                            log.error("", e);
                        }
                    }
                }

                //log.error(name + "--" + needCheckHeartbeat + "-" + isStopped() + "--执行完成了---------------------------------------------------------------------------------------------------执行完成了");
            }
        }, "tio-timer-checkheartbeat-" + id + "-" + name);
        checkHeartbeatThread.setDaemon(true);
        checkHeartbeatThread.setPriority(Thread.MIN_PRIORITY);
        checkHeartbeatThread.start();
    }

    private boolean onHeartbeatTimeout(ChannelContext channelContext, long interval, int incrementAndGet) {
        return true;
    }


    @Override
    public Handler getTioHandler() {
        return this.getServerHandler();
    }


    public ServerHandler getServerHandler() {
        return serverHandler;
    }


    @Override
    public boolean isServer() {
        return true;
    }


}
